/**
 * Copyright (C) 2017-2018 pinganfang, Inc. All Rights Reserved.
 */
package com.pingan.haofang.searchcloud.register.metadata;

import com.pingan.haofang.framework.common.functional.lang.data.Try;
import com.pingan.haofang.searchcloud.api.matedata.FieldMeta;
import com.pingan.haofang.searchcloud.api.matedata.IndexMeta;
import com.pingan.haofang.searchcloud.api.matedata.IndexMetadataHolder;
import com.pingan.haofang.searchcloud.api.matedata.MetaCacheSupport;
import com.pingan.haofang.searchcloud.common.exception.SystemException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import java.util.Map;
import java.util.Objects;
import java.util.Optional;

/**
 * @author huhailiang937
 * @date 2017年6月16日下午5:01:09
 * @since 1.0.0
 */
@Service("redisIndexMetadataRegister")
@ManagedResource
public class RedisIndexMetadataRegister implements IndexMetadataHolder {

    private static RedisTemplate redisTemplate;

    protected static Logger LOG = LoggerFactory.getLogger(RedisIndexMetadataRegister.class);

    // JDK ServiceLoader 初始化需无参构造器
    public RedisIndexMetadataRegister() {
        LOG.info("-----> JDK ServiceLoader  IndexMetadataHolder for redisIndexMetadataRegister init sucess");
    }

    @Autowired
    public RedisIndexMetadataRegister(ApplicationContext applicationContext) {
        redisTemplate = (RedisTemplate) applicationContext.getBean("redisQueueTemplate");
        if(redisTemplate == null){
            LOG.error("redisIndexMetadataRegister init error ,can not found redisTemplate");
            throw new RuntimeException("redisIndexMetadataRegister init error ,can not found redisTemplate");
        }
    }


    @Override
    public Optional<FieldMeta> findFiledMeta(String indexName, String fieldName) {
        Objects.requireNonNull(indexName, "indexName is null");
        Objects.requireNonNull(fieldName, "fieldName is null");
        return Optional.ofNullable(findIndexMeta(indexName).getFiledMetas().get(fieldName));
    }

    @Override
    public Optional<String> findPKFieldName(String indexName) {
        Objects.requireNonNull(indexName, "indexName is null");
        Map<String, FieldMeta> filedMetaMap = Try.of(() -> findIndexMeta(indexName).getFiledMetas())
                .orElseThrow(RuntimeException::new);
        for (Map.Entry<String, FieldMeta> keyValue : filedMetaMap.entrySet()) {
            String key = keyValue.getKey();
            FieldMeta fieldMeta = keyValue.getValue();
            if (fieldMeta.isPk()) {
                return Optional.of(key);
            }
        }
        return Optional.empty();
    }

    @Override
    public IndexMeta findIndexMeta(String indexName) {
        Objects.requireNonNull(indexName, "indexName is null");
        try {
            String json = (String) redisTemplate.opsForHash().get(MetaCacheSupport.ROOT_META_KEY, indexName);
            if (json == null) {
                return null;
            }
            try {
                IndexMeta metaInZk = MetaCacheSupport.decode4Json(json, IndexMeta.class);
                return metaInZk;
            } catch (Exception e) {
                LOG.error(String.format("redisIndexMetadataRegister discover(%s) has error: ", indexName), e);
                return null;
            }
        } catch (Exception e) {
            throw new SystemException("search.indexname.notexist", new String[]{indexName});
        }
    }

    @Override
    public boolean register(IndexMeta indexMeta) {
        Assert.notNull(indexMeta, "indexMeta can not be null");
        Assert.notNull(indexMeta.getIndexName(), "indexMeta can not be null");
        String json = null;
        try {
            json = new String(MetaCacheSupport.encode(indexMeta));
            redisTemplate.opsForHash().put(MetaCacheSupport.ROOT_META_KEY, indexMeta.getIndexName(), json);
            LOG.info(String.format("redisIndexMetadataRegister register Meta : %s", json));
            return true;
        } catch (Exception e) {
            LOG.error(String.format("redisIndexMetadataRegister register fail! setData(%s) has error:", json), e);
        }
        return false;
    }

}
